package org.apache.rocketmq.store.queue;



import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.attribute.CQType;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.QueueTypeUtils;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.store.*;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;

import static java.lang.String.format;
import static org.apache.rocketmq.store.config.StorePathConfigHelper.getStorePathBatchConsumeQueue;
import static org.apache.rocketmq.store.config.StorePathConfigHelper.getStorePathConsumeQueue;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/29
 * @方法描述：消费队列管理器，在第十四版本代码中，这个类中的内容非常简单
 */
public class ConsumeQueueStore extends AbstractConsumeQueueStore {


    public ConsumeQueueStore(DefaultMessageStore messageStore) {
        super(messageStore);
    }



    @Override
    public void start() {
        log.info("Default ConsumeQueueStore start!");
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载消费队列内存映射文件的方法
     */
    @Override
    public boolean load() {
        //加载普通消费队列的操作
        boolean cqLoadResult = loadConsumeQueues(getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()), CQType.SimpleCQ);
        //加载批量消费队列的操作，程序中并没有用到批量消费队列，所以并不会真的执行加载批量消费队列的操作
        boolean bcqLoadResult = loadConsumeQueues(getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir()), CQType.BatchCQ);
        //返回加载结果
        return cqLoadResult && bcqLoadResult;
    }


    //销毁后重新加载消费队列的方法
    @Override
    public boolean loadAfterDestroy() {
        return true;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：程序重启时恢复数据的方法，在第十五版本代码中并没有用到这个方法，十六版本才会用到，所以我就先不添加注释了
     */
    @Override
    public void recover() {
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueueInterface logic : maps.values()) {
                this.recover(logic);
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：并发恢复消费队列方法，在第十五版本代码中并没有用到这个方法，所以我就先不添加注释了
     */
    @Override
    public boolean recoverConcurrently() {
        int count = 0;
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            count += maps.values().size();
        }
        final CountDownLatch countDownLatch = new CountDownLatch(count);
        BlockingQueue<Runnable> recoverQueue = new LinkedBlockingQueue<>();
        final ExecutorService executor = buildExecutorService(recoverQueue, "RecoverConsumeQueueThread_");
        List<FutureTask<Boolean>> result = new ArrayList<>(count);
        try {
            for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
                for (final ConsumeQueueInterface logic : maps.values()) {
                    FutureTask<Boolean> futureTask = new FutureTask<>(() -> {
                        boolean ret = true;
                        try {
                            logic.recover();
                        } catch (Throwable e) {
                            ret = false;
                            log.error("Exception occurs while recover consume queue concurrently, " +
                                    "topic={}, queueId={}", logic.getTopic(), logic.getQueueId(), e);
                        } finally {
                            countDownLatch.countDown();
                        }
                        return ret;
                    });
                    result.add(futureTask);
                    executor.submit(futureTask);
                }
            }
            countDownLatch.await();
            for (FutureTask<Boolean> task : result) {
                if (task != null && task.isDone()) {
                    if (!task.get()) {
                        return false;
                    }
                }
            }
        } catch (Exception e) {
            log.error("Exception occurs while recover consume queue concurrently", e);
            return false;
        } finally {
            executor.shutdown();
        }
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：关闭程序的方法
     */
    @Override
    public boolean shutdown() {
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：计算下一个内存映射文件起始偏移量的方法
     */
    @Override
    public long rollNextFile(ConsumeQueueInterface consumeQueue, final long offset) {
        //根据消费队列主题和Id找到对应的文件队列生命周期管理器，这里我解释以下，这个文件队列生命周期管理器实际上就是ConsumeQueue对象
        //因为ConsumeQueue类实现了FileQueueLifeCycle接口
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        //计算下一个内存映射文件的起始偏移量
        return fileQueueLifeCycle.rollNextFile(offset);
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：修正指定消费队列最小偏移量的方法
     */
    public void correctMinOffset(ConsumeQueueInterface consumeQueue, long minCommitLogOffset) {
        consumeQueue.correctMinOffset(minCommitLogOffset);
    }

    /**
     * 分发消息
     */
    @Override
    public void putMessagePositionInfoWrapper(DispatchRequest dispatchRequest){
        ConsumeQueueInterface cq = this.findOrCreateConsumeQueue(dispatchRequest.getTopic(), dispatchRequest.getQueueId());
        //调用ConsumeQueue对象的putMessagePositionInfoWrapper方法
        this.putMessagePositionInfoWrapper(cq, dispatchRequest);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：查找指定消费队列中指定范围的消息数据的方法
     */
    @Override
    public List<ByteBuffer> rangeQuery(String topic, int queueId, long startIndex, int num) {
        //直接返回null，表示不支持范围查询功能
        return null;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到指定消费队列指定偏移量的消息数据的方法
     */
    @Override
    public ByteBuffer get(String topic, int queueId, long startIndex) {
        //直接返回null，表示不支持该功能
        return null;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到指定消费队列中最大消息偏移量的方法
     */
    @Override
    public long getMaxOffsetInQueue(String topic, int queueId) {
        //根据主题和队列Id找到目标消费队列
        ConsumeQueueInterface logic = findOrCreateConsumeQueue(topic, queueId);
        if (logic != null) {
            //如果队列存在则直接获取该队列的最大消息偏移量
            return logic.getMaxOffsetInQueue();
        }
        //队列不存在则返回0
        return 0;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据消息存储时间戳查找指定消费队列的消息偏移量的方法，该方法在第十五版本中并没有被用到，我就先不添加注释了
     */
    @Override
    public long getOffsetInQueueByTime(String topic, int queueId, long timestamp, BoundaryType boundaryType) {
        ConsumeQueueInterface logic = findOrCreateConsumeQueue(topic, queueId);
        if (logic != null) {
            long resultOffset = logic.getOffsetInQueueByTime(timestamp, boundaryType);
            resultOffset = Math.max(resultOffset, logic.getMinOffsetInQueue());
            resultOffset = Math.min(resultOffset, logic.getMaxOffsetInQueue());
            return resultOffset;
        }
        return 0;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到文件队列生命周期管理器的方法
     */
    private FileQueueLifeCycle getLifeCycle(String topic, int queueId) {
        return findOrCreateConsumeQueue(topic, queueId);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载指定消费队列的方法，其实就是加载该消费队列的内存映射文件
     */
    public boolean load(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        return fileQueueLifeCycle.load();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：加载指定路径下的消费队列的方法，
     */
    private boolean loadConsumeQueues(String storePath, CQType cqType) {
        //得到存储消费队列文件的目录的File对象
        File dirLogic = new File(storePath);
        //得到目录下所有文件的数组，注意，这里得到的其实还是文件夹，只不过文件夹的名称是主题信息
        //"D:/code/store/consumequeue/TestTopic/2/00000000000000000000"这是消费队列文件的存储路径
        //现在得到的这个数组存储的就是主题文件夹
        File[] fileTopicList = dirLogic.listFiles();
        if (fileTopicList != null) {
            //遍历啊主题文件夹
            for (File fileTopic : fileTopicList) {
                //得到消费队列存储的消息所属的主题
                String topic = fileTopic.getName();
                //得到主题文件夹下的所有目录文件，这个时候得到的就是消费队列的队列Id文件夹
                File[] fileQueueIdList = fileTopic.listFiles();
                if (fileQueueIdList != null) {
                    //判断是否得到了队列Id文件夹
                    for (File fileQueueId : fileQueueIdList) {
                        //得到了则开始遍历队列Id文件夹
                        int queueId;
                        try {
                            //得到当前遍历到的消费队列文件夹的队列Id
                            queueId = Integer.parseInt(fileQueueId.getName());
                        } catch (NumberFormatException e) {
                            continue;
                        }
                        //查询当前遍历到的队列的类型是否正确，也就是看看队列类型是否和主题配置信息对应
                        queueTypeShouldBe(topic, cqType);
                        //为当前遍历到的消费队列创建对应的消费队列对象，在第十五版本代码中创建的就是ConsumeQueue对象
                        ConsumeQueueInterface logic = createConsumeQueueByType(cqType, topic, queueId, storePath);
                        //把创建完毕的消费队列对象存放到父类的consumeQueueTable成员变量中
                        this.putConsumeQueue(topic, queueId, logic);
                        //在这里执行加载消费队列内存映射文件的操作
                        if (!this.load(logic)) {
                            //如果任何一个队列加载失败，就立即返回false
                            return false;
                        }
                    }
                }
            }
        }
        log.info("load {} all over, OK", cqType);
        return true;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据消费队列类型创建对应的消费队列对象的方法
     */
    private ConsumeQueueInterface createConsumeQueueByType(CQType cqType, String topic, int queueId, String storePath) {
        //如果是简单类型则创建ConsumeQueue对象
        if (Objects.equals(CQType.SimpleCQ, cqType)) {
            return new ConsumeQueue(
                    topic,
                    queueId,
                    storePath,
                    this.messageStoreConfig.getMappedFileSizeConsumeQueue(),
                    this.messageStore);
        }
        //如果是批量类型则创建BatchConsumeQueue对象
        else if (Objects.equals(CQType.BatchCQ, cqType)) {
            return new BatchConsumeQueue(
                    topic,
                    queueId,
                    storePath,
                    this.messageStoreConfig.getMapperFileSizeBatchConsumeQueue(),
                    this.messageStore);
        }
        else {
            //未知类型则直接抛出异常
            throw new RuntimeException(format("queue type %s is not supported.", cqType.toString()));
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：校验消费队列类型和主题一致性的方法
     */
    private void queueTypeShouldBe(String topic, CQType cqTypeExpected) {
        //得到该主题的配置信息
        Optional<TopicConfig> topicConfig = this.messageStore.getTopicConfig(topic);
        //从配置信息中得到该主题消费队列的类型
        CQType cqTypeActual = QueueTypeUtils.getCQType(topicConfig);
        //判断实际队列类型是否与期望队列类型一致
        if (!Objects.equals(cqTypeExpected, cqTypeActual)) {
            //类型不一致则抛出异常
            throw new RuntimeException(format("The queue type of topic: %s should be %s, but is %s", topic, cqTypeExpected, cqTypeActual));
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：构建执行器的方法，在执行消费队列信息恢复的操作时，会把操作交给当前方法创建的执行器来执行，但在第十五版本代码中还用不上这个方法
     * 第十六版本就会用上了
     */
    private ExecutorService buildExecutorService(BlockingQueue<Runnable> blockingQueue, String threadNamePrefix) {
        return ThreadUtils.newThreadPoolExecutor(
                this.messageStore.getBrokerConfig().getRecoverThreadPoolNums(),
                this.messageStore.getBrokerConfig().getRecoverThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                blockingQueue,
                new ThreadFactoryImpl(threadNamePrefix));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：程序重启时恢复指定消费队列信息的方法
     */
    public void recover(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.recover();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到消费队列中的最大物理偏移量的方法
     */
    @Override
    public Long getMaxPhyOffsetInConsumeQueue(String topic, int queueId) {
        ConsumeQueueInterface logic = findOrCreateConsumeQueue(topic, queueId);
        if (logic != null) {
            return logic.getMaxPhysicOffset();
        }
        return null;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到所有消费队列中最大的消息偏移量的方法
     */
    @Override
    public long getMaxPhyOffsetInConsumeQueue() {
        long maxPhysicOffset = -1L;
        //遍历所有的消费队列
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueueInterface logic : maps.values()) {
                //判读哪个队列的最大的消息偏移量最大
                if (logic.getMaxPhysicOffset() > maxPhysicOffset) {
                    //得到最新的最大偏移量
                    maxPhysicOffset = logic.getMaxPhysicOffset();
                }
            }
        }
        return maxPhysicOffset;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到指定消费队列中最小偏移量的方法
     */
    @Override
    public long getMinOffsetInQueue(String topic, int queueId) {
        ConsumeQueueInterface logic = findOrCreateConsumeQueue(topic, queueId);
        if (logic != null) {
            return logic.getMinOffsetInQueue();
        }
        return -1;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费队列自检的方法
     */
    public void checkSelf(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.checkSelf();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：所有消费队列自检的方法，也就是判断所有消费队列的数据完整性的方法
     */
    @Override
    public void checkSelf() {
        for (Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> topicEntry : this.consumeQueueTable.entrySet()) {
            for (Map.Entry<Integer, ConsumeQueueInterface> cqEntry : topicEntry.getValue().entrySet()) {
                this.checkSelf(cqEntry.getValue());
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费队列数据刷盘的方法
     */
    @Override
    public boolean flush(ConsumeQueueInterface consumeQueue, int flushLeastPages) {
        //得到制定消费队列
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        //在这里执行实际刷盘操作
        return fileQueueLifeCycle.flush(flushLeastPages);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：销毁指定消费队列的方法
     */
    @Override
    public void destroy(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.destroy();
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除指定消费队列过期文件的方法，该方法在第十五版本代码中还用不到，第十六版本代码就会用到了
     */
    @Override
    public int deleteExpiredFile(ConsumeQueueInterface consumeQueue, long minCommitLogPos) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        return fileQueueLifeCycle.deleteExpiredFile(minCommitLogPos);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：阶段消费队列脏数据的方法，该方法在第十五版本代码中还用不到，第十六版本代码就会用到了
     */
    public void truncateDirtyLogicFiles(ConsumeQueueInterface consumeQueue, long phyOffset) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.truncateDirtyLogicFiles(phyOffset);
    }



    //交换内存的方法
    public void swapMap(ConsumeQueueInterface consumeQueue, int reserveNum, long forceSwapIntervalMs, long normalSwapIntervalMs) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.swapMap(reserveNum, forceSwapIntervalMs, normalSwapIntervalMs);
    }


    //清理以交换的映射内存的方法
    public void cleanSwappedMap(ConsumeQueueInterface consumeQueue, long forceCleanSwapIntervalMs) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        fileQueueLifeCycle.cleanSwappedMap(forceCleanSwapIntervalMs);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断消费队列的第一个内存映射文件是否可用的方法
     */
    @Override
    public boolean isFirstFileAvailable(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        return fileQueueLifeCycle.isFirstFileAvailable();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断消费队列的第一个内存映射文件是否存在的方法
     */
    @Override
    public boolean isFirstFileExist(ConsumeQueueInterface consumeQueue) {
        FileQueueLifeCycle fileQueueLifeCycle = getLifeCycle(consumeQueue.getTopic(), consumeQueue.getQueueId());
        return fileQueueLifeCycle.isFirstFileExist();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：根据主题和队列Id查找目标消费队列的方法
     */
    @Override
    public ConsumeQueueInterface findOrCreateConsumeQueue(String topic, int queueId) {
        //获得对应主题的队列映射表
        ConcurrentMap<Integer, ConsumeQueueInterface> map = this.consumeQueueTable.get(topic);
        if (map == null){
            ConcurrentMap<Integer, ConsumeQueueInterface> newMap = new ConcurrentHashMap<>();
            ConcurrentMap<Integer, ConsumeQueueInterface> oldMap = this.consumeQueueTable.putIfAbsent(topic, newMap);
            if (oldMap != null) {
                map = oldMap;
            }else{
                map = newMap;
            }
        }
        //获得对应队列Id的消费队列
        ConsumeQueueInterface logic = map.get(queueId);
        if (logic != null) {
            return logic;
        }

        //不存在则意味着要创建新的消费队列
        ConsumeQueueInterface newLogic;
        //得到当前主题的配置信息，从配置信息中可以得到该主题对应的消费队列的具体类型
        Optional<TopicConfig> topicConfig = this.messageStore.getTopicConfig(topic);
        // TODO maybe the topic has been deleted.
        //根据消费队列类型创建对应的消费队列
        if (Objects.equals(CQType.BatchCQ, QueueTypeUtils.getCQType(topicConfig))) {
            //如果是批量消费队列类型，创建BatchConsumeQueue对象
            newLogic = new BatchConsumeQueue(
                    topic,
                    queueId,
                    getStorePathBatchConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
                    this.messageStoreConfig.getMapperFileSizeBatchConsumeQueue(),
                    this.messageStore);
        } else {
            //否则创建普通消费队列对象
            newLogic = new ConsumeQueue(
                    topic,
                    queueId,
                    getStorePathConsumeQueue(this.messageStoreConfig.getStorePathRootDir()),
                    this.messageStoreConfig.getMappedFileSizeConsumeQueue(),
                    this.messageStore);
        }
        //将新创建的消费队列放入映射表
        ConsumeQueueInterface oldLogic = map.putIfAbsent(queueId, newLogic);
        //确定最终返回的消费队列
        if (oldLogic != null) {
            //如果已存在消费队列则使用已经存在的
            logic = oldLogic;
        } else {
            //否则使用新创建的消费队列
            logic = newLogic;
        }
        //返回找到或创建的消费队列
        return logic;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：设置批量主题队列映射表的方法
     */
    public void setBatchTopicQueueTable(ConcurrentMap<String, Long> batchTopicQueueTable) {
        this.queueOffsetOperator.setBatchTopicQueueTable(batchTopicQueueTable);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：更新指定消费队列数据偏移量的方法
     */
    public void updateQueueOffset(String topic, int queueId, long offset) {
        //得到指定消费队列的键
        String topicQueueKey = topic + "-" + queueId;
        //更新消费队列数据偏移量
        this.queueOffsetOperator.updateQueueOffset(topicQueueKey, offset);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：将消费队列存放到父类的consumeQueueTable消费队列映射表中的方法
     */
    private void putConsumeQueue(final String topic, final int queueId, final ConsumeQueueInterface consumeQueue) {
        //从父类的consumeQueueTable映射表中中获取主题对应的消费队列映射表，表的key是队列Id，value是消费队列本身
        ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface> map = this.consumeQueueTable.get(topic);
        if (null == map) {
            //如果映射表不存在则创建
            map = new ConcurrentHashMap<>();
            //然后将指定队列存放到映射表中
            map.put(queueId, consumeQueue);
            //再把新创建的映射表放到consumeQueueTable表中
            this.consumeQueueTable.put(topic, map);
        } else {
            //如果映射表存在，则直接把消费队列放到表中即可
            map.put(queueId, consumeQueue);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法在第十五版本代码中还用不到，暂时不添加注释
     */
    @Override
    public void recoverOffsetTable(long minPhyOffset) {
        ConcurrentMap<String, Long> cqOffsetTable = new ConcurrentHashMap<>(1024);
        ConcurrentMap<String, Long> bcqOffsetTable = new ConcurrentHashMap<>(1024);
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueueInterface logic : maps.values()) {
                String key = logic.getTopic() + "-" + logic.getQueueId();
                long maxOffsetInQueue = logic.getMaxOffsetInQueue();
                if (Objects.equals(CQType.BatchCQ, logic.getCQType())) {
                    bcqOffsetTable.put(key, maxOffsetInQueue);
                } else {
                    cqOffsetTable.put(key, maxOffsetInQueue);
                }
                this.correctMinOffset(logic, minPhyOffset);
            }
        }
        if (messageStoreConfig.isDuplicationEnable() || messageStore.getBrokerConfig().isEnableControllerMode()) {
            compensateForHA(cqOffsetTable);
        }
        this.setTopicQueueTable(cqOffsetTable);
        this.setBatchTopicQueueTable(bcqOffsetTable);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法在第十五版本代码中还用不到，暂时不添加注释
     */
    private void compensateForHA(ConcurrentMap<String, Long> cqOffsetTable) {
        SelectMappedBufferResult lastBuffer = null;
        long startReadOffset = messageStore.getCommitLog().getConfirmOffset() == -1 ? 0 : messageStore.getCommitLog().getConfirmOffset();
        log.info("Correct unsubmitted offset...StartReadOffset = {}", startReadOffset);
        while ((lastBuffer = messageStore.selectOneMessageByOffset(startReadOffset)) != null) {
            try {
                if (lastBuffer.getStartOffset() > startReadOffset) {
                    startReadOffset = lastBuffer.getStartOffset();
                    continue;
                }
                ByteBuffer bb = lastBuffer.getByteBuffer();
                int magicCode = bb.getInt(bb.position() + 4);
                if (magicCode == CommitLog.BLANK_MAGIC_CODE) {
                    startReadOffset += bb.getInt(bb.position());
                    continue;
                }
                else if (magicCode != MessageDecoder.MESSAGE_MAGIC_CODE) {
                    throw new RuntimeException("Unknown magicCode: " + magicCode);
                }
                lastBuffer.getByteBuffer().mark();
                DispatchRequest dispatchRequest = messageStore.getCommitLog().checkMessageAndReturnSize(lastBuffer.getByteBuffer(), true, messageStoreConfig.isDuplicationEnable(), true);
                if (!dispatchRequest.isSuccess())
                    break;
                lastBuffer.getByteBuffer().reset();
                MessageExt msg = MessageDecoder.decode(lastBuffer.getByteBuffer(), true, false, false, false, true);
                if (msg == null)
                    break;
                String key = msg.getTopic() + "-" + msg.getQueueId();
                cqOffsetTable.put(key, msg.getQueueOffset() + 1);
                startReadOffset += msg.getStoreSize();
                log.info("Correcting. Key:{}, start read Offset: {}", key, startReadOffset);
            } finally {
                if (lastBuffer != null)
                    lastBuffer.release();
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：销毁所有消费队列的方法
     */
    @Override
    public void destroy() {
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueueInterface logic : maps.values()) {
                this.destroy(logic);
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：删除消费队列过期文件的方法，该方法在第十五版本代码中还用不到，第十六版本代码就会用到了
     */
    @Override
    public void cleanExpired(long minCommitLogOffset) {
        Iterator<Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>>> it = this.consumeQueueTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, ConcurrentMap<Integer, ConsumeQueueInterface>> next = it.next();
            String topic = next.getKey();
            if (!TopicValidator.isSystemTopic(topic)) {
                ConcurrentMap<Integer, ConsumeQueueInterface> queueTable = next.getValue();
                Iterator<Map.Entry<Integer, ConsumeQueueInterface>> itQT = queueTable.entrySet().iterator();
                while (itQT.hasNext()) {
                    Map.Entry<Integer, ConsumeQueueInterface> nextQT = itQT.next();
                    long maxCLOffsetInConsumeQueue = nextQT.getValue().getLastOffset();
                    if (maxCLOffsetInConsumeQueue == -1) {
                        log.warn("maybe ConsumeQueue was created just now. topic={} queueId={} maxPhysicOffset={} minLogicOffset={}.",
                                nextQT.getValue().getTopic(),
                                nextQT.getValue().getQueueId(),
                                nextQT.getValue().getMaxPhysicOffset(),
                                nextQT.getValue().getMinLogicOffset());
                    } else if (maxCLOffsetInConsumeQueue < minCommitLogOffset) {
                        log.info(
                                "cleanExpiredConsumerQueue: {} {} consumer queue destroyed, minCommitLogOffset: {} maxCLOffsetInConsumeQueue: {}",
                                topic,
                                nextQT.getKey(),
                                minCommitLogOffset,
                                maxCLOffsetInConsumeQueue);
                        removeTopicQueueTable(nextQT.getValue().getTopic(),
                                nextQT.getValue().getQueueId());
                        this.destroy(nextQT.getValue());
                        itQT.remove();
                    }
                }
                if (queueTable.isEmpty()) {
                    log.info("cleanExpiredConsumerQueue: {},topic destroyed", topic);
                    it.remove();
                }
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：阶段消费队列脏数据的方法，该方法在第十五版本代码中还用不到，第十六版本代码就会用到了
     */
    @Override
    public void truncateDirty(long offsetToTruncate) {
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            for (ConsumeQueueInterface logic : maps.values()) {
                this.truncateDirtyLogicFiles(logic, offsetToTruncate);
            }
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到所有消费队列文件中存储的消息总大小的方法
     */
    @Override
    public long getTotalSize() {
        long totalSize = 0;
        //遍历所有主题的消费队列
        for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : this.consumeQueueTable.values()) {
            //遍历每个主题下的所有队列
            for (ConsumeQueueInterface logic : maps.values()) {
                //累加每个队列的存储的数据大小
                totalSize += logic.getTotalSize();
            }
        }
        //返回计算得到的数据总大小
        return totalSize;
    }
}
